package com.peter.springcloud.streamclient.sink;

import java.util.Date;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;

import com.peter.springcloud.streamclient.client.RabbitClient;

/**
 * 消息驱动的一个实现。
 * @author peter
 */
@EnableBinding(RabbitClient.class)
public class SinkReceiver {
    private static Logger logger = LogManager.getLogger(SinkReceiver.class);
    private ThreadPoolExecutor executor;
    private final LinkedBlockingQueue<Runnable> taskQueue;
    private final static int TASK_QUEUE_CAPACITY = 1000; //30;
    private int count = 0;

    public SinkReceiver() {
        // executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(10);
        // 限定对了容量：30
        taskQueue = new LinkedBlockingQueue<Runnable>(TASK_QUEUE_CAPACITY);
        executor = new ThreadPoolExecutor(10, 10, 0, TimeUnit.SECONDS, taskQueue);
    }

    /**
     * 订阅接收消息。
     */
    @StreamListener(RabbitClient.INPUT)
    // @SendTo("responseMessage") //该注解会在消息处理完成后，向responseMessage这个队列发送消息。消息内容就是该方法的返回值。
    public void receive(String payload) {
        while(executor.getQueue().size() >= TASK_QUEUE_CAPACITY) {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        try {

            Future<Boolean> future = executor.submit(new ServerCall(payload));
            logger.info("添加任务完成：" + count++ + ", " + executor.getQueue().size() + ", " + executor.getTaskCount());
            if(future.isDone()) {
                logger.info("这么快就完成了！！！");
            }
        }
        catch(RejectedExecutionException e) {
           logger.error("Exception: " + e.getClass().getName() + ", " + e.getMessage());
           // 如果线程池用完了，休息两秒钟 —— 正常时候应该不会走这里，因为
           try {
            Thread.sleep(5000);
        } catch (InterruptedException e1) {
            e1.printStackTrace();
        }
           throw e;
        }
    }

    public class ServerCall implements Callable<Boolean> {
        private Object payload;

        public ServerCall(Object payload) {
            this.payload = payload;
        }

        @Override
        public Boolean call() throws Exception {
            long time = new Date().getTime();
            logger.info("Received: " + payload + ", start of " + time);
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                logger.info("InterruptedException: " + e.getMessage() + " of " + time);
                return false;
            }
            finally {
                logger.info("Received: " + payload + ", end of " + time);
            }
            return true;
        }

    }
}
